#ifndef DATAFLOW_DETAIL_SCHEDULER_H
#define DATAFLOW_DETAIL_SCHEDULER_H

#include <assert.h>

#include <queue>

#include <QObject>
#include <QMutex>
#include <QMutexLocker>
#include <QWaitCondition>

#include "Graph.h"

#include "Node.h"
#include "NodePtrVector.h"

#include "TaskInstance.h"
#include "TaskInstancePtrList.h"

namespace DataFlow
{

namespace Detail
{

class Scheduler : public QObject
{
	Q_OBJECT

	public:

		typedef QObject   BaseType;
		typedef Scheduler ThisType;

		Scheduler(const Graph & g, QObject * parent = 0)
			: BaseType(parent)
                        , finishedFlag(false)
                        , tasksEnabled(true)
		{
			this->connect
			(
				this,
				SIGNAL(internalTryScheduleTaskSignal(int)),
				SLOT(internalTryScheduleTaskSlot(int)),
				Qt::QueuedConnection
			);

			this->connect
			(
				this,
				SIGNAL(internalTaskStartedSignal(DataFlow::Detail::TaskInstance *)),
				SLOT(internalTaskStartedSlot(DataFlow::Detail::TaskInstance *)),
				Qt::QueuedConnection
			);

			this->connect
			(
				this,
				SIGNAL(internalTaskFinishedSignal(DataFlow::Detail::TaskInstance *)),
				SLOT(internalTaskFinishedSlot(DataFlow::Detail::TaskInstance *)),
				Qt::QueuedConnection
			);

			this->runningInstanceInvalidIterator = this->runningInstances.end();

			this->setGraph(g);
		}

		virtual ~Scheduler(void)
		{
			for (size_t i=0; i<this->nodes.size(); ++i)
			{
				delete this->nodes[i];
			}
		}

		void dispatchTasks(bool on)
		{
			if (this->tasksEnabled != on)
			{
				this->tasksEnabled = on;
				this->waitCond.wakeAll();
			}
		}

		void pushInput(int task, int port, int id, const Argument & arg)
		{
			this->finishedFlag = false;
			assert((task >= 0) && (task < int(this->nodes.size())));
			Node * n = this->nodes[task];
			assert((port >= 0) && (port < n->inputSize()));
			n->pushInput(port, id, arg);
			emit internalTryScheduleTaskSignal(task);
		}

		TaskInstance * nextTaskInstance(void)
		{
			TaskInstance * t = 0;

			this->mutex.lock();
			while (this->readyInstances.empty() && this->tasksEnabled)
			{
				this->waitCond.wait(&(this->mutex));
			}
			if (!this->readyInstances.empty())
			{
				t = this->readyInstances.top();
				this->readyInstances.pop();
			}
			this->mutex.unlock();

			if (t == 0) return 0;

			const bool r = t->node->getAvailableInputById(t->input);
			assert(r);
			(void)r;

			emit internalTaskStartedSignal(t);
			return t;
		}

		void taskInstanceFinished(TaskInstance * t)
		{
			assert(t != 0);
			t->node->feedOutput(t->output);
			emit internalTaskFinishedSignal(t);
		}

		void waitForFinished(void)
		{
			this->executionMutex.lock();
			while (!this->finishedFlag)
			{
				this->executionCond.wait(&(this->executionMutex));
			}
			this->executionMutex.unlock();
		}

		/*
		void waitForCanceled(void)
		{
			;
		}

		void waitForPaused(void)
		{
			;
		}

		void waitForResumed(void)
		{
			;
		}
		*/

	public slots:

		/*
		void cancel(void)
		{
			this->mutex.lock();
			const bool wasSet = this->setExecutionFlags(ThisType::SF_CANCEL);
			this->mutex.unlock();
			if (!wasSet) { this->executionCond.wakeAll(); }
		}

		void pause(void)
		{
			this->mutex.lock();
			const bool wasSet = this->setExecutionFlags(ThisType::SF_PAUSE);
			this->mutex.unlock();
			if (!wasSet) { this->executionCond.wakeAll(); }
		}

		void resume(void)
		{
			QMutexLocker locker(&this->mutex);
			if (!this->doPause) return;
			if (this->doResume) return;
			this->doResume = true;
		}
		*/

	signals:

		/*
		void canceled     (void);
		void paused       (void);
		void resumed      (void);
		*/

		void taskStarted  (int task, DataFlow::ArgumentArray input);
		void taskFinished (int task, DataFlow::ArgumentArray input, DataFlow::ArgumentArray output);
		void finished     (void);

		void internalTryScheduleTaskSignal  (int task);
		void internalTaskStartedSignal      (DataFlow::Detail::TaskInstance * t);
		void internalTaskFinishedSignal     (DataFlow::Detail::TaskInstance * t);

	protected:

		struct TaskInstanceCompare
		{
			bool operator () (const TaskInstance * lhs, const TaskInstance * rhs) const
			{
				assert(lhs != 0);
				assert(rhs != 0);
				if (lhs->node->id() != rhs->node->id()) return (lhs->node->id() < rhs->node->id());
				return (lhs->input.id() > rhs->input.id());
			}
		};

		typedef std::priority_queue<TaskInstance *,  std::vector<TaskInstance *>, TaskInstanceCompare> TaskInstancePriorityQueue;

		QMutex                      mutex;
		QWaitCondition              waitCond;
		TaskInstancePtrList         runningInstances;
		TaskInstancePtrListIterator runningInstanceInvalidIterator;
		TaskInstancePriorityQueue   readyInstances;
		NodePtrVector               nodes;

		QMutex         executionMutex;
		QWaitCondition executionCond;
		bool           finishedFlag;

		bool tasksEnabled;

		void tryScheduleNodes(const NodePtrVector & nodes)
		{
			for (size_t i=0; i<nodes.size(); ++i)
			{
				Node * n = nodes[i];
				if (n == 0) continue;
				emit internalTryScheduleTaskSignal(n->id());
			}
		}

		void setGraph(const Graph & g)
		{
			const GraphNodeVector & gnodes = g.nodes();
			this->nodes.resize(gnodes.size());

			for (size_t i=0; i<this->nodes.size(); ++i)
			{
				this->nodes[i] = new Node(int(i));
			}

			const GraphLinkSet & glinks = g.links();
			for (GraphLinkSetConstIterator it=glinks.begin(); it!=glinks.end(); ++it)
			{
				const GraphLink & gl = (*it);

				Node * sourceNode = 0;
				if (gl.sourceNode >= 0) sourceNode = this->nodes[gl.sourceNode];

				Node * targetNode = 0;
				if (gl.targetNode >= 0) targetNode = this->nodes[gl.targetNode];

				if (sourceNode != 0) sourceNode->outputLinks().push_back(NodeOutputLinkDescriptor(gl.sourcePort, targetNode, gl.targetPort, gl.capacity));
				if (targetNode != 0) targetNode->inputLinks().push_back(NodeInputLinkDescriptor(sourceNode, gl.sourcePort, gl.targetPort, gl.capacity));
			}

			for (size_t i=0; i<gnodes.size(); ++i)
			{
				const GraphNode & gn = gnodes[i];
				this->nodes[i]->setup(gn.task, gn.inputSize, gn.outputSize);
			}
		}

	protected slots:

		void internalTryScheduleTaskSlot(int task)
		{
			assert((task >= 0) && (task < int(this->nodes.size())));

			Node * n = this->nodes[task];

			std::vector<TaskInstance *> instances;
			while (true)
			{
				int id = 0;
				if (!n->prepareToRun(id)) break;

				TaskInstance * t = new TaskInstance(n->task(), n->id(), n, id, n->inputSize(), n->outputSize(), this->runningInstanceInvalidIterator);
				instances.push_back(t);
			}

			bool noMoreData = false;

			this->mutex.lock();
			if (!instances.empty())
			{
				for (size_t i=0; i<instances.size(); ++i)
				{
					this->readyInstances.push(instances[i]);
				}
			}
			else
			{
				if (this->readyInstances.empty() && this->runningInstances.empty())
				{
					noMoreData = true;
					for (size_t i=0; i<this->nodes.size(); ++i)
					{
						if (!this->nodes[i]->inputIsEmpty())
						{
							noMoreData = false;
							break;
						}
					}
				}
			}
			this->mutex.unlock();

			if (!instances.empty())
			{
				this->waitCond.wakeAll();
			}

			if (noMoreData)
			{
				this->finishedFlag = true;
				this->executionCond.wakeAll();
				emit finished();
			}
		}

		void internalTaskStartedSlot(DataFlow::Detail::TaskInstance * t)
		{
			{
				assert(t != 0);
				QMutexLocker locker(&this->mutex);
				assert(t->runningListIterator == this->runningInstanceInvalidIterator);
				t->runningListIterator = this->runningInstances.insert(this->runningInstances.end(), t);
			}

			Node * n = t->node;
			this->tryScheduleNodes(n->inputNodes());
		}

		void internalTaskFinishedSlot(DataFlow::Detail::TaskInstance * t)
		{
			Node * n = t->node;

			{
				assert(t != 0);
				QMutexLocker locker(&this->mutex);
				TaskInstancePtrListIterator it = t->runningListIterator;
				assert(it != this->runningInstanceInvalidIterator);
				t->runningListIterator = this->runningInstanceInvalidIterator;
				this->runningInstances.erase(it);
				delete t;
			}

			this->tryScheduleNodes(n->outputNodes());

			emit internalTryScheduleTaskSignal(n->id());
		}
};

}

}

#endif // DATAFLOW_DETAIL_SCHEDULER_H
